package vn.com.epi.re;

import java.io.Closeable;
import java.io.IOException;
import java.io.PrintStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;

import org.apache.commons.cli.CommandLine;
import org.apache.commons.cli.HelpFormatter;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
import org.apache.commons.cli.PosixParser;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import vn.com.epi.re.tools.HTableWrapper;


import com.google.common.collect.Lists;

/**
 * <p>
 * Hold object shared across the application. Usage: whenever you write a main
 * class, include this snippet:
 * 
 * <pre>
 * ApplicationContext c = ApplicationContext.init(); 
 * try {
 *     ... // all codes go here
 * } finally {
 *     Closeables.closeQuietly(c);
 * }
 * </pre>
 * 
 * While you write an unit test, include these code:
 * 
 * <pre>
 * &#064;BeforeClass
 * public static void setUpClass() {
 * 	ApplicationContext.init();
 * }
 * 
 * &#064;AfterClass
 * public static void tearDownClass() {
 * 	if (ApplicationContext.get() != null) {
 * 		ApplicationContext.get().close();
 * 	}
 * }
 * </pre>
 * 
 * </p>
 * <p>
 * <strong>Caution!</strong> Don't use this class in mappers/reducers! Because
 * each of them run in a separate JVM, they don't automatically have an instance
 * of application context and by no means they can obtain necessary information
 * to initialize one.
 * </p>
 * 
 */
public class ApplicationContext implements Closeable {

    private static final Log LOG = LogFactory.getLog(ApplicationContext.class);
    
	private Configuration conf;
	private HTablePool pool;
	private FileSystem fs;
	
	public ApplicationContext(Configuration conf) throws IOException {
		this.conf = conf;
		fs = FileSystem.get(conf);
	}
	
	public Configuration getConf() {
		return conf;
	}

	public synchronized HTablePool getTablePool() {
	    if (pool == null) {
	        pool = new HTablePool(conf, 10);
        }
		return pool;
	}

	/**
	 * Utility method to get an instance of HTableInterface. The same as
	 * appContext.getTablePool().getTable(name).
	 * 
	 * @param name
	 * @return a table object
	 */
	public HTableWrapper getTable(byte[] name) {
		return new HTableWrapper(getTablePool().getTable(name));
	}
	
	/**
	 * Utility method to get an instance of HTableInterface. The same as
	 * appContext.getTablePool().getTable(name).
	 * 
	 * @param name
	 * @return a table object
	 */
	public HTableWrapper getTable(String name) {
		return new HTableWrapper(getTablePool().getTable(name));
	}

	/**
	 * Utility to create FileSystem object with current configuration. The same
	 * as FileSystem.get(appContext.getConf()).
	 * 
	 * @return file system object
	 */
	public FileSystem getFileSystem() {
		return fs;
	}

	/**
	 * Run a tool with current configuration. Replace {@link ToolRunner#run(Tool, String[])}.
	 * 
	 * @param tool
	 * @param args
	 * @return
	 * @throws Exception
	 */
    public int run(Tool tool, String... args) throws Exception {
        GenericOptionsParser parser = new GenericOptionsParser(conf, args);
        String[] toolArgs = parseApplicationOptions(conf, parser.getRemainingArgs());
        tool.setConf(conf);
        fs = FileSystem.get(conf);
        LOG.info("Zookeeper address: " + conf.get("hbase.zookeeper.quorum"));
        LOG.info("Zookeeper port: " + conf.get("hbase.zookeeper.property.clientPort"));
        return tool.run(toolArgs);
    }

    /**
     * This method add JAR files to distributed cache but doesn't update
     * client's classpath (because it's impossible). Download client library to
     * Hadoop library folder is a better way that will work for sure.
     * 
     * @param conf
     * @param dir
     * @throws IOException
     * @throws URISyntaxException
     */
    public static void addDependenciesFromDirectory(Configuration conf,
            String dir) throws IOException, URISyntaxException {
        List<Path> addedPaths = Lists.newLinkedList();
        FileSystem fs = FileSystem.get(new URI(dir), conf);
        for (FileStatus fileStatus : fs.listStatus(new Path(dir))) {
            DistributedCache.addFileToClassPath(fileStatus.getPath(), conf, fs);
            addedPaths.add(fileStatus.getPath());
        }
        LOG.info("Files added to distributed cache and local class path: " + addedPaths);
        DistributedCache.createSymlink(conf);
    }

    private static String[] parseApplicationOptions(Configuration conf,
            String[] args) throws ParseException, IOException {
        Options options = new Options();
        options.addOption("libdir", true, "A directory consisting JAR files of dependencies.");
        CommandLine cmd = new PosixParser().parse(options, args, true);
        if (cmd.hasOption("libdir")) {
            try {
                addDependenciesFromDirectory(conf, cmd.getOptionValue("libdir"));
            } catch (URISyntaxException e) {
                throw new IOException(e);
            }
        }
        return cmd.getArgs();
    }
    
    public static void printApplicationOptions(PrintStream out) {
        out.print("epi-tre generic options:");
        out.println("hbaseconf\tPath to HBase configuration to be used.");
    }
    
	@Override
	public void close() throws IOException {
		if (pool != null) {
			pool.close();
		}
	}

	public static void formatHelp(String syntax, Options options) {
        new HelpFormatter().printHelp(syntax, options, true);
        GenericOptionsParser.printGenericCommandUsage(System.out);
        printApplicationOptions(System.out);
    }

    private static final Object INIT_LOCK = new Object();
	private static ApplicationContext instance;

	/**
	 * FIXME: how to get conf properly?
	 * @throws IOException 
	 */
	public static ApplicationContext init() throws IOException {
		return init(new Configuration());
	}

	public static ApplicationContext init(Configuration conf) throws IOException {
		synchronized (INIT_LOCK) {
			if (instance != null) {
				throw new IllegalStateException(
						"Application context can be initialized only once.");
			}
			instance = new ApplicationContext(conf);
			return instance;
		}
	}
	
	public static ApplicationContext get() {
		return instance;
	}
	
}
